package com.shujia.core

import com.shujia.core.Demo10Join.Student
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo13ReduceByKey {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo13ReduceByKey")
    conf.setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    // 读取学生数据及分数数据 并将每一行数据转换成样例类对象
    val stuRDD: RDD[Student] = sc
      .textFile("Spark/data/students.txt")
      .map(line => {
        val splits: Array[String] = line.split(",")
        val id: String = splits(0)
        val name: String = splits(1)
        val age: Int = splits(2).toInt
        val gender: String = splits(3)
        val clazz: String = splits(4)
        Student(id, name, age, gender, clazz)
      })
    // 统计班级人数

    // 将stuRDD变成KV格式 使用groupByKey进行分组
    stuRDD
      .map(stu => (stu.clazz, 1))
      .groupByKey()
      .map(kv => s"${kv._1},${kv._2.size}")
      .foreach(println)

    /**
     * 在计算过程中 Shuffle阶段是最消耗时间的
     *
     * reduceByKey 转换算子
     * 需要接收一个 聚合操作（只能用sum、count、max、min）
     * 首先按照Key进行分组 会在 组内 进行 聚合操作
     * 然后 也会在 组之间 进行聚合操作
     * 相当于MR中的Combiner预聚合操作
     *
     * 相比groupByKey可以减少Shuffle过程中传输的数据量 进而提高运行效率
     *
     */
    stuRDD
      .map(stu => (stu.clazz, 1))
      //      .reduceByKey((i, j) => i + j)
      .reduceByKey(_ + _) // 简写
      .foreach(println)

    while (true) {

    }
  }

  /**
   * SparkPi是怎么实现的？
   */

}
